
----------------------------------------------------------------
--- Copyright：  迷你玩-海马
--- Description：各业务服务器连接到data_cache缓存服务器时使用，
--- 使用ngx自带的连接池，长连接到data_cache并发送读写请求，
---
--- 业务服务器 --(ssdbDcClient.lua)--> data_cache缓存服务器 ----> SSDB数据库
----------------------------------------------------------------

local assert     = assert
local require    = require
local bit        = bit
local coroutine  = coroutine
local debug      = debug
local io         = io
local pairs      = pairs
local ipairs     = ipairs
local math       = math
local os         = os
local print      = print
local pcall      = pcall
local xpcall     = xpcall
local rawget     = rawget
local rawset     = rawset
local select     = select
local string     = string
local table      = table
local tonumber   = tonumber
local tostring	 = tostring
local error      = error
local type       = type
local unpack     = unpack
local setmetatable = setmetatable
local getmetatable = getmetatable
local ngx = ngx

require "resty.core.shdict"

local ngx_shared  = ngx.shared;
local string_sub  = string.sub
local string_gsub = string.gsub
local string_format = string.format

local log        = require("modules.commhm.log");
local ns_time    = require("modules.commhm.time");
local uu         = require("modules.commhm.utils");
local ns_network = require("modules.commhm.network");

local ns_centerServerProxy   = require("modules.commhm.centerServerProxy");

local table_insert = table.insert
local ngx_sleep = ngx.sleep
local ngx_now   = ngx.now


-------------------------------------------------------------------------------
--  __READ_ONLY__  只读   不用锁
-------------------------------------------------------------------------------


-- SSDB MGR类
local _M = { _VERSION = '0.12.1' }
local mt = { __index = _M }


---构造函数，使用配置文件初始化一个data_cache连接
function _M.new(self, module_name_, config_server_, config_select_ )
	log.debug( log.fast() and "call ssdbDcClient:new" );
	local ins_ = {
		config_cache  = {};
		module_name   = module_name_;

		config_server = config_server_;
		config_select = config_select_;

		tj = {};   --统计模块

		carry_index = ngx.worker.id() * 10000 + 10000;
	};

	-- 异步上报config，失败不影响使用
	local config_str_ = table.tostring( { server=config_server_, select=config_select_ } )
	ns_centerServerProxy.WWW_configReport( module_name_ .. '_dc', config_str_ )

    return setmetatable(ins_, mt);
end



---释放当次请求中的所有远程锁， 一般在 gg.final() 中调用
---扫描当前请求中，哪些地方使用了远程锁，并全部释放掉
function _M.final( self )

	--释放locks
	if  ngx.ctx.resty_dc_locks then

		log.debug( log.fast() and "try unlock all resty_dc_locks ssdb" )
		uu.var_dump( ngx.ctx.resty_dc_locks )

		for k, v in pairs( ngx.ctx.resty_dc_locks ) do
			if  v and v.select then
				if  v.cc > 0 then
					local ok, err_ = self:send_unlock_kh_to_dc( k, v.select )  --远程解锁
					if  ok then
						if  v.cc > 1 then
							log.day_list_simple( "ssdb", "lock_rw", "unlock_ok2", ngx.worker.id(),
									uu.get_act_cmd(), k, v.cc, debug.traceback() );
						end
					else
						log.day_list_simple( "ssdb", "lock_rw", "error_unlock2", ngx.worker.id(),
								uu.get_act_cmd(), k, v.cc, err_, debug.traceback() );
					end
				else
					--log.day_list_simple( "ssdb", "lock_rw", "unlocked2", ngx.worker.id(),
						--uu.get_act_cmd(), k, v.cc, debug.traceback() );
				end
			else
				log.day_list_simple( "ssdb", "lock_rw", "error_locks2", ngx.worker.id(),
						uu.get_act_cmd(), k, v.cc, debug.traceback() );
			end
		end
		ngx.ctx.resty_dc_locks = nil
	end

end



---解锁全部远程锁，请求过程中途主动释放，很少有地方使用
function _M.unlock_all( self )
	self:final()
end


---解锁指定kh远程锁
function _M.unlock_key_hash( self, select_, key_, hash_ )
	local key_hash_ = key_ .. "_h" .. hash_
	self:send_unlock_kh_to_dc( key_hash_, select_ )
end


------------------------------------
---批量读取 (不会对数据加锁)
function _M.mread( self, select_, kv_list_, err_call_back )
	log.debug( log.fast() and "call ssdbDcClient:mread" );

	local kv_list_t_ = {}
	for i=1, #kv_list_ do
		kv_list_t_[ #kv_list_t_ + 1 ] =  kv_list_[i].key
		kv_list_t_[ #kv_list_t_ + 1 ] =  kv_list_[i].hash
	end

	if  #kv_list_t_ <= 0 then
		return {}
	end

	local kv_list_str_ = table.concat( kv_list_t_, '-' )

	local pp_ = {
		func    = 'mread',
		select  = select_,
		kv_list = kv_list_str_,
	}

	local ret_ = self:private_post_to_dc_server( pp_, err_call_back )
	local t_ = table.loadstring( ret_ ) or {}

	if  type(t_) == 'table' then
		for i=1, #t_ do
			if  t_[i] == "not_found" then
				t_[i] = ngx.null           --与原版本一致
			end
		end
	end

	return t_
end


----主动对数据进行加远程锁 (基本上不需要调用)
-- 一般read操作的时候会加锁， save或者final解锁
-- 特殊情况下需要直接加远程锁，目前只有ma服务在使用
function _M.add_read_lock( self, select_, key, hash )

	local key_hash_ = key .. "_h" .. hash

	if  ngx.ctx.__READ_ONLY__ == 1 then
		---不用加锁
	else
		---是否已经被锁住了，如果已经锁了
		if  not ngx.ctx.resty_dc_locks then
			ngx.ctx.resty_dc_locks = {}
		end

		if  ngx.ctx.resty_dc_locks[ key_hash_ ] then
			if  ngx.ctx.resty_dc_locks[ key_hash_ ].cc > 0 then
				--命中了cache，可以重复锁定
			else
				ngx.ctx.resty_dc_locks[ key_hash_ ].cc = ngx.ctx.resty_dc_locks[ key_hash_ ].cc + 1
			end
		else
			--新锁定成功
			ngx.ctx.resty_dc_locks [ key_hash_ ] = { cc=1, select=select_ }
			--log.day_list_simple( "ssdb", "lock_rw", "lock_ok",  ngx.worker.id(), uu.get_act_cmd(), 0, key_hash_ );
		end

		if  log.fast() then
			log.debug( "add resty_dc_locks1 : " .. key_hash_ )
			uu.var_dump( ngx.ctx.resty_dc_locks )
		end

	end
end



---读取单个数据，可选加远程锁
---select:  用来做十分表或者百分表的分表id，一般是uin或者地图id
---key:     一般是uin或者地图id
---hash:    同一个uin下的分类，参考 const_vars.lua 里的 FOO.SSDB.KEYS
---err_call_back:  出错回调
---no_cache_  :  1=不使用缓存          其他(默认值)=使用缓存
---read_only_ :  1=只读，不加远程锁    其他(默认值)=加远程锁
---如果设置了 ngx.ctx.__READ_ONLY__ = 1， 表示整个请求中，所有read操作都不加锁
function _M.read( self, select_, key, hash, err_call_back, no_cache_, read_only_ )
	local pp_ = {
		func    = 'read',
		select  = select_,
		key     = key,
		hash    = hash,
	}

	if  no_cache_ == 'y' then
		pp_.no_cache = 'y'
	end

	local key_hash_ = key .. "_h" .. hash

	if  (ngx.ctx.__READ_ONLY__ == 1) or (read_only_ == 1) then
		log.debug( log.fast() and "add resty_dc_locks : read_only ignore" )
		pp_.read_only = 1   --只读
	else
		---是否已经被自己锁住了，如果已经锁了，则不用再加锁，转为一个只读请求
		if  not ngx.ctx.resty_dc_locks then
			ngx.ctx.resty_dc_locks = {}
		end

		if  ngx.ctx.resty_dc_locks[ key_hash_ ] and ngx.ctx.resty_dc_locks[ key_hash_ ].cc > 0 then
			ngx.ctx.resty_dc_locks[ key_hash_ ].cc = ngx.ctx.resty_dc_locks[ key_hash_ ].cc + 1
			pp_.read_only = 1   --只读
			log.day_list_simple( "ssdb", "lock_rw", "relock_add_0", ngx.worker.id(), uu.get_act_cmd(), key_hash_,
					ngx.ctx.resty_dc_locks[ key_hash_ ].cc, debug.traceback() );
		end

		log.debug( log.fast() and "add resty_dc_locks2 : " .. key_hash_ )
		uu.var_dump( ngx.ctx.resty_dc_locks )
	end

	local ret_ = self:private_post_to_dc_server( pp_, err_call_back, 3000 )

	if  ret_ == '' then
		---返回空 (一般是等待获取远程锁失败) 重试
		if  ngx.ctx.__no_Content_Length__ then
			log.error( "private_post_to_dc_server retry " .. uu.get_act_cmd() ..  "\n" .. key_hash_ )
			ret_ = self:private_post_to_dc_server( pp_, err_call_back )
		else
			log.error( "private_post_to_dc_server has content-length, ignore retry " .. key_hash_ )
		end
	end

	if  ret_  and ( not (pp_.read_only == 1) ) then
		--记录key_hash, 被锁定了
		if  not ngx.ctx.resty_dc_locks then
			ngx.ctx.resty_dc_locks = {}
		end

		if  ngx.ctx.resty_dc_locks[ key_hash_ ] then
			ngx.ctx.resty_dc_locks[ key_hash_ ].cc = ngx.ctx.resty_dc_locks[ key_hash_ ].cc + 1

			if  ngx.ctx.resty_dc_locks[ key_hash_ ].cc > 1 then
				--重复锁定
				log.day_list_simple( "ssdb", "lock_rw", "relock_add_1", ngx.worker.id(), uu.get_act_cmd(), key_hash_,
						ngx.ctx.resty_dc_locks[ key_hash_ ].cc, debug.traceback() );
			end

			log.debug( log.fast() and "add resty_dc_locks3 : " .. key_hash_ )
			uu.var_dump( ngx.ctx.resty_dc_locks )
		else
			--新锁成功
			ngx.ctx.resty_dc_locks [ key_hash_ ] = { cc=1, select=select_ }
			--log.day_list_simple( "ssdb", "lock_rw", "lock_ok",  ngx.worker.id(), uu.get_act_cmd(), 0, key_hash_ );

			log.debug( log.fast() and "add resty_dc_locks4 : " .. key_hash_ )
			uu.var_dump( ngx.ctx.resty_dc_locks )
		end
	end

	if  ret_ == "not_found" then
		ret_ = ngx.null          --与原版本纯ssdb接口一致
	end

	return ret_
end



---写入单个数据，并解开远程锁
---select:  用来做十分表或者百分表的分表id，一般是uin
---key:     一般是uin
---hash:    同一个uin下的分类，参考 const_vars.lua 里的FOO.SSDB.KEYS
---err_call_back:  出错回调
---delay_       :  1=不立即写ssdb    其他(默认值)=立即同时写ssdb
---read_only_   :  1=不解开远程锁    其他(默认值)=如果有远程锁，立即解锁
-----             （可作为主动参数传入=1，暂时不用解锁，等待final解锁，适用于一次读取，多次写入的情况）
function _M.save( self, select_, key, hash, value, err_call_back, delay_, read_only_ )
	local pp_ = {
		func    = 'save',
		select  = select_,
		key     = key,
		hash    = hash,
		value   = value,
		read_only  = read_only_,  --是否不解远程锁
	}

	if  delay_ == 1 then
		pp_.ssdb_delay = 1  --是否不立即写ssdb
	end

	if  ngx.ctx.__READ_ONLY__ == 1 then
		--只读接口有写操作
		log.day_list_simple( "ssdb", "lock_rw", "error_save_where_read_only", ngx.worker.id(),
				uu.get_act_cmd(), key .. "_h" .. hash, debug.traceback() );
	end

	--写操作 判断一下锁
	local ret_ = self:private_post_to_dc_server( pp_, err_call_back, 3000 )
	if  ret_ then
		--试图解锁
		local key_hash_ = key .. "_h" .. hash
		if  read_only_ == 1 then
			--不用解锁 ( 或者作为主动参数传入，暂时不用解锁，等待final解锁 )
		elseif  ngx.ctx.resty_dc_locks and ngx.ctx.resty_dc_locks[ key_hash_ ] then
			if  ngx.ctx.resty_dc_locks[ key_hash_ ].cc > 0 then
				ngx.ctx.resty_dc_locks[ key_hash_ ].cc = 0   --解锁成功
				--log.day_list_simple( "ssdb", "lock_rw", "unlock_ok",
					--ngx.worker.id(), uu.get_act_cmd(), 0, key_hash_ )
			else
				--重复解锁 使用缓存连写两次
				--log.day_list_simple( "ssdb", "lock_rw", "warn_re_unlock", ngx.worker.id(),
						--uu.get_act_cmd(), 0, key_hash_, debug.traceback() );
			end
		else
			--找不到锁
			log.day_list_simple( "ssdb", "lock_rw", "error_not_find_lock", ngx.worker.id(),
					    uu.get_act_cmd(), 'nil', key_hash_, debug.traceback() );
		end
	end
	return ret_
end


--- 删除一个key下的一个hash，也删除对应的value
---[key][hash1]=[value1]
function _M.del( self, select, key, hash, err_call_back )
	local pp_ = {
		func    = 'del',
		select  = select,
		key     = key,
		hash    = hash,
	}
	return self:private_post_to_dc_server( pp_, err_call_back )
end


--- 获得一个key下的所有hash值
---[key][hash1]=[value1]
---[key][hash2]=[value2]
---[key][hash3]=[value3]
function _M.hlist( self, select, start_, end_, size_, err_call_back)
	local pp_ = {
		func    = 'hlist',
		select  = select,
		start   = start_,
		h_end   = end_,
		size    = size_,
	}
	return self:private_post_to_dc_server( pp_, err_call_back )
end


--- 获得一个key下有多少key hash组合
-----[key][hash1]=[value1]
-----[key][hash2]=[value2]
-----[key][hash3]=[value3]
function _M.hgetall( self, select, key_, err_call_back)
	local pp_ = {
		func    = 'hget_all',
		select  = select,
		key     = key_,
	}
	return self:private_post_to_dc_server( pp_, err_call_back )
end


--- 一个key下有多少个hash值
function _M.hlen( self, select, key, err_call_back)
	local pp_ = {
		func    = 'hlen',
		select  = select,
		key     = key,
	}
	return self:private_post_to_dc_server( pp_, err_call_back )
end


---获得一个key下的所有值
-----[key][hash1]=[value1]
-----[key][hash2]=[value2]
-----[key][hash3]=[value3]
function _M.hget_all( self, select, key, err_call_back )
	local pp_ = {
		func    = 'hget_all',
		select  = select,
		key     = key,
	}
	return self:private_post_to_dc_server( pp_, err_call_back )
end


---设置单个kv
function _M.set_kv( self, select, key, value, et, err_call_back )
	local pp_ = {
		func    = 'set_kv',
		select  = select,
		key     = key,
		value   = value,
		et      = et,
	}
	return self:private_post_to_dc_server( pp_, err_call_back )
end


---获得单个kv
function _M.get_kv( self, select, key, err_call_back )
	local pp_ = {
		func    = 'get_kv',
		select  = select,
		key     = key,
	}
	local ret_ = self:private_post_to_dc_server( pp_, err_call_back )
	if  ret_ == "not_found" then
		ret_ = ngx.null          --与原版本一致
	end
	return ret_
end


---发送远程解锁请求，解锁一个key hash对
---(一般情况下read加锁，save解锁，或者 final() 统一解锁，很少这样去主动解锁)
function _M.send_unlock_kh_to_dc( self, key_hash_, select_, err_call_back )
	local pp_ = {
		func    = 'unlock_kh',
		select  = select_,
		kh      = key_hash_,
	}
	return self:private_post_to_dc_server( pp_, err_call_back, 3000 )
end


---远程清理缓存， 后续的读请求会重新加载ssdb数据
function _M.clean_shared_cache_ssdb(self, select_, key_, hash_ )
	local pp_ = {
		func    = 'clean_cache',
		select  = select_,
		key     = key_,
		hash    = hash_,
	}
	return self:private_post_to_dc_server( pp_, 3000 )
end



--------------------------------private functions--------------------------
---- 业务应该不会使用到下面的函数 ----

--- 发送请求到服务器 POST
function _M.private_post_to_dc_server( self, pp_, err_call_back, p_time_exp_ )

	ngx.ctx.err_call_back = err_call_back;
	local url_ = self:private_get_url_by_select( pp_.select )

	self.carry_index = self.carry_index + 1
	local Carry_ = '' .. self.carry_index

	--http://120.24.64.132:8087/dc/?auth=Xcn67K&name=posting&cmd=comm&name=map&carry=111aaa
	url_ = url_ .. "&cmd=comm&carry=" .. Carry_ .. "&select=" .. pp_.select
	if  ngx.ctx.m_params.uin then
		url_ = url_ .. "&uin=" .. ngx.ctx.m_params.uin
	end

	local post_data_ = ngx.encode_args(pp_);
	local ret_, body_, resp_ = ns_network.postHttpPage( url_, post_data_, p_time_exp_ )

	if  log.fast() then
		local url_pos_ = string.find( url_, '?' )
		if  url_pos_ and url_pos_>0 then
			log.debug( "call private_post_to_dc_server: " .. string.sub( url_, 1, url_pos_-1 ) )
		else
			log.debug( "call private_post_to_dc_server: " .. url_ )
		end
		uu.var_dump( pp_ )
		log.debug( post_data_ )
		log.debug( "ssdb post ret=" .. ret_ .. ", body=" .. (body_ or 'nil') )
	end

	if  ret_ == 200 then
		--验证Carry
		if  resp_.headers.Carry == Carry_ then
			if  resp_.headers["Content-Length"] then
				if  resp_.headers["Content-Length"] ==  ('' .. #body_) then
					return body_   --ok
				else
					log.day_list_simple( "ssdb", "dc_client", "DataLen_check_error",
							resp_.headers["Content-Length"], #body_ )
				end
			else
				ngx.ctx.__no_Content_Length__ = 1
				return body_       --ok 没有Content-Length
			end
		else
			log.day_list_simple( "ssdb", "dc_client", "Carry_check_error", resp_.headers.Carry, #Carry_ )
		end
	else
		log.day_list_simple( "ssdb", "dc_client", "http_error", ret_, url_ )
	end

	self:error_exit( 'error select = ' .. (pp_.select or 'nil') );
end


---------------------------------------
---根据ssdb_name获得连接的url
function _M.private_get_url_by_select( self, select_ )
	if  self.config_cache[ select_ ]  then
		--先查config cache
		log.debug( log.fast() and "find config_cache for " .. select_ );
		return  self.config_cache[ select_ ];
	else
		if  self.config_server and self.config_select then
			if  self.config_select[select_] and
				self.config_server[ self.config_select[select_] ]     and
				self.config_server[ self.config_select[select_] ].url then
				local ret_ = self.config_server[ self.config_select[select_] ].url;
				if  ret_ then
					---log.debug( log.fast() and "set config_cache for " .. select_ );
					self.config_cache[ select_ ] = ret_;
					return ret_;
				end
			end
		end
	end

	self:error_exit( 'error ssdb_name = ' .. (select_ or 'nil') );

end



-- 出错处理
function _M.error_exit(self, txt)
	if  ngx.ctx.err_call_back then
		log.error( "has set error_exit." );
		ngx.ctx.err_call_back(txt);
	else
		log.error( "not set error_exit." );
	end
end



---------------------------------------------校验函数----------------------------
--写校验[_k_] [_h_]
_M.private_check_set_key = function(self, select_, key, hash, value )
	if  value._k_ then
		if  ( tonumber(value._k_) and ( tonumber(key) == tonumber(value._k_) ) ) or (key == value._k_) then
			--ok
		else
			log.day_list_short( "ssdb", "error_key_check", "set", select_, key, value._k_, hash, debug.traceback() );
			log.error( "error_set_key_check" )
			return false;
		end
	else
		value._k_ = tonumber(key) or key;  --首次设置
	end

	if  value._h_ then
		if  ( tonumber(value._h_) and ( tonumber(hash) == tonumber(value._h_) ) ) or (hash == value._h_) then
			--ok
		else
			log.day_list_short( "ssdb", "error_hash_check", "set", select_, key, value._h_, hash );
			log.error( "error_set_hash_check" )
			return false;
		end
	else
		if  tonumber(hash) then
			value._h_ = tonumber(hash) --首次设置
		end
	end

	value._t_ = uu.now()    --最后一次写入时间
	return true;
end



--读校验[_k_] [_h_]
_M.private_check_get_key = function(self, select_, key, hash, value )
	if  value._k_ then
		if  ( tonumber(value._k_) and ( tonumber(key) == tonumber(value._k_) ) ) or (key == value._k_) then
			--ok
		else
			--读取出错
			log.day_list_short( "ssdb", "error_key_check", "get", select_, key, value._k_, hash, debug.traceback() );
			log.error( "error_get_key_check" )
			return false;
		end
	end

	if  value._h_ then
		if  ( tonumber(value._h_) and ( tonumber(hash) == tonumber(value._h_) ) ) or (hash == value._h_) then
			--ok
		else
			--读取出错
			log.day_list_short( "ssdb", "error_hash_check", "get", select_, key, value._h_, hash );
			log.error( "error_get_hash_check" )
			return false;
		end
	end

	return true;
end



return _M;
